package cz.cvut.fel.persister;

import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.HashMap;

import javax.mail.internet.NewsAddress;

import org.bson.BSON;
import org.bson.BSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.basho.riak.pbc.RequestMeta;
import com.basho.riak.pbc.RiakClient;
import com.basho.riak.pbc.RiakObject;
import com.mongodb.BasicDBObject;

import cz.cvut.fel.email.Email;
import cz.cvut.fel.loader.Persister;

public class RiakPersister extends Persister {
	final static Logger log = LoggerFactory.getLogger(RiakPersister.class);
	public static final String BUCKET_ATTACHMENTS = "ATTACHMENTS";
	public static final String BUCKET_DOMAINA = "DOMAINA";
	
	RiakClient riakClient;
	
	
	public RiakPersister(String server, Integer port) {
		super(parseServer(server), (port == null ? parsePort(server) : port));
	}

	@Override
	public void connect() throws Exception {
			riakClient = new RiakClient(server);
	}

	@Override
	public void store(Email email) throws Exception {
		ArrayList<String> keys = new ArrayList<String>();
	//	log.info(email.getHeaders().toString());
		for (int i = 0; i < email.getAttachments().size(); i++) {
			try {
				RequestMeta meta = new RequestMeta();
				meta.w(2);
				keys.add(email.getKeys().get(i));
				RiakObject obj	= new RiakObject(BUCKET_ATTACHMENTS,email.getKeys().get(i), email.getAttachments().get(i).getBytes());				
				riakClient.store(obj,meta);
				//log.info("STORED "+email.getKeys().get(i));
			} catch (Exception e) {
				log.error("Unable to store email " + email, e);

			}
		}
		
		if (email.getContent() != null) {
			try {
				HashMap<String,String> headers = email.getHeaders();
				String mailKey = URLEncoder.encode(email.getHeaders().get("MESSAGE-ID"),"UTF-8");
				BasicDBObject bson = new BasicDBObject ();

				try {
					bson.put("FROM", headers.get("X-VF-Scanner-Mail-From".toUpperCase()).split(" ")[0]);
				} catch(Exception e) {
					bson.put("FROM", "dummy@dummy.com");
				}
				try {
					bson.put("MAIL-TO", headers.get("X-VF-Scanner-Rcpt-To".toUpperCase()).split(",")[0]);
				} catch(Exception e) {
					bson.put("MAIL-TO", "dummy@dummy.com");
				}
				
				bson.put("DATE_date", headers.get("DATE"));
				bson.put("SUBJECT", headers.get("SUBJECT"));
				bson.put("DATA", email.getContent());
				//bson.toString().length();
				RiakObject obj	= new RiakObject("DOMAIN",mailKey,  bson.toString());
				obj.setContentType("text/x-json");
				
				for (String key : keys) {
					obj.addLink("ATTACHMENTS", BUCKET_ATTACHMENTS, key);	
				}
				
				riakClient.store(obj);
				
				
				
				//log.info("STORED PRIMARY KEY="+mailKey+" ATT COUNT="+email.getAttachments().size());
			} catch (Exception e) {
				log.error("Unable to store email " + email, e);

			}
		}
		
	//	log.info("MID: "+email.getHeaders().get("MESSAGE-ID").length() + ", FROM: " + email.getHeaders().get("FROM").length()+ " , TO: "+email.getHeaders().get("TO").length()+ " , DATE: "+email.getHeaders().get("DATE").length()+ " , SUBJ: "+email.getHeaders().get("SUBJECT").length());
	
	}



	@Override
	public void disconnect() {
		riakClient = null;

	}

	
	public Email readEmail() throws Exception {
		return null;
		// TODO Auto-generated method stub
		
	}

	@Override
	public byte[] readBytes() throws Exception {
		// TODO Auto-generated method stub
		return null;
	}



}
